package com.mutouren.common.mq.kafka;

import java.io.Closeable;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import com.mutouren.common.log.LogAlias;
import com.mutouren.common.log.LogManager;
import com.mutouren.common.log.Logger;

public class NewConsumer<K, V> implements Closeable{
	
	private Properties props;
	private KafkaConsumer<K, V> consumer;
	private ExecutorService executor = Executors.newCachedThreadPool();
	private volatile boolean isOpened = false;
	
	protected final static Logger errorLogger = LogManager.getLogger(LogAlias.SystemError.name());
	
	@SuppressWarnings("unchecked")
	public NewConsumer(String bootstrapServers, String groupId, Class<V> valueType) {
		this(bootstrapServers, groupId, false, (Class<K>)Integer.class, valueType);
	}
	
	public NewConsumer(String bootstrapServers, String groupId, boolean isSmallestOffset,
			Class<K> keyType, Class<V> valueType) {
		this(getProperties(bootstrapServers, groupId, isSmallestOffset, keyType, valueType));
	}
	
	public NewConsumer(Properties props) {
		this.props = props;
		this.consumer = new KafkaConsumer<>(this.props);
	}
	
	private static Properties getProperties(String bootstrapServers, String groupId, boolean isSmallestOffset,
			Class<?> keyType, Class<?> valueType) {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("group.id", groupId);
        if (isSmallestOffset) {
        	props.put("auto.offset.reset", "earliest");
        } else {
        	props.put("auto.offset.reset", "largest");
        }
		props.put("key.deserializer", getDeserializer(keyType));
		props.put("value.deserializer", getDeserializer(valueType));

//        props.put("enable.auto.commit", "true");
//        props.put("auto.commit.interval.ms", "1000");
//        props.put("session.timeout.ms", "30000");
   
		return props;
	}
	
	public static String getDeserializer(Class<?> clazz) {
		String result;
		
		if (clazz== Integer.class) {
			result = "org.apache.kafka.common.serialization.IntegerDeserializer";
		} else if (clazz == Long.class) {
			result = "org.apache.kafka.common.serialization.LongDeserializer";
		} else if (clazz == byte[].class) {	
			result = "org.apache.kafka.common.serialization.ByteArrayDeserializer";
		} else if (clazz == String.class) {
			result = "org.apache.kafka.common.serialization.StringDeserializer";
		} else {
			throw new IllegalArgumentException( "consumer deserializer don't know the type: " + clazz);
		}		
		return result;
	}	
	
	public void addListener (final String topic, final NewConsumerCallback<K, V> newCallback) {
		this.isOpened = true;
		this.consumer.subscribe(Arrays.asList(topic));
		this.executor.submit(new Runnable() {

			@Override
			public void run() {
				while(isOpened) {
					ConsumerRecords<K, V> records = NewConsumer.this.consumer.poll(100);
					for (ConsumerRecord<K, V> record : records) {
						//display(record);
						newCallback.callback(record);
					}
				}				
			}			
		});
	}
	
	public void display(ConsumerRecord<K, V> record) {
		System.out.println(String.format("%s: partition[%d], offset[%d], key=%s, value=%s", 
				Thread.currentThread().getName(),
				record.partition(),
				record.offset(),
				record.key(),
				record.value()
				));		
	}
	
	public static interface NewConsumerCallback<K, V> {
		void callback(ConsumerRecord<K, V> record);
	}

	@Override
	public void close() {
		this.isOpened = false;
		this.executor.shutdownNow();
		try {
			executor.awaitTermination(10, TimeUnit.SECONDS);	
		} catch (Throwable t) {
			errorLogger.error("NewConsumer close error", t);
		}		
		this.consumer.close();		
	}	

}
